let amqp = require('amqplib');

class RabbitMQ {
    constructor() {
        this.hosts = ["amqp://localhost"];
        this.index = 0;
        this.length = this.hosts.length;
    }
    receiveQueueMsg(queueName, receiveCallBack, errCallBack) {
        let self = this;

        amqp.connect(this.hosts[this.index])
            .then(function (conn) {
                // console.log("receive connect success")
                return conn.createChannel()
                    .then(function (channel) {
                        // console.log("receive createChannel success")
                        channel.on('close', function() {
                            // console.log("receive connection closed")
                            conn.close();
                        })
                        return channel.assertQueue(queueName, { durable: true })
                            .then(function (ok) {
                                // console.log("receive assertQueue success")
                                channel.prefetch(1); // 当有一个任务未完成时，不再接受新任务
                                return channel.consume(queueName, function (msg) {
                                    // console.log("consume msg:" + msg)
                                    if (msg !== null) {
                                        let data = msg.content.toString();
                                        channel.ack(msg);
                                        receiveCallBack && receiveCallBack(data);
                                    }
                                }, { noAck: false })
                                    .finally(function () {
                                        setTimeout(() => {
                                            if (channel) {
                                                // console.log("receive channel closed")
                                                channel.close();
                                            }
                                        }, 500)
                                    });
                            })
                    })
                    .catch((error) => {
                        console.log("receive createChannel error : " + error)
                    })
            })
            .catch(function () {
                if (self.index <= self.length - 2) {
                    self.index++
                } else {
                    self.index = 0;
                }
            });
    }
    sendQueueMsg(queueName, msg, errCallBack) {
        let self = this;

        amqp.connect(this.hosts[this.index])
            .then(function (conn) {
                // console.log("send connect success")
                return conn.createChannel()
                    .then(function (channel) {
                        // console.log("send createChannel success")
                        channel.on('close', function() {
                            // console.log("send connection closed")
                            conn.close();
                        })
                        return channel.assertQueue(queueName, { durable: true })
                            .then(function (ok) {
                                // console.log("send assertQueue success")
                                return channel.sendToQueue(queueName, Buffer.from(msg), {
                                    persistent: true,
                                    mandatory: true
                                });
                            })
                            .then(function (data) {
                                if (data) {
                                    // console.log("send: " + JSON.stringify(data) )
                                    errCallBack && errCallBack("success");

                                    // console.log("send channel closed")
                                    channel.close();
                                }
                            })
                            .catch(function () {
                                setTimeout(() => {
                                    if (channel) {
                                        console.log("send channel closed (error)")
                                        channel.close();
                                    }
                                }, 500)
                            });
                    })
            })
            .catch(function () {
                if (self.index <= self.length - 2) {
                    self.index++
                } else {
                    self.index = 0;
                }
            });
    }
}

// let mq = new RabbitMQ();
// mq.receiveQueueMsg('testQueue', (msg) => {
//     console.log(msg)//123
// })

// mq.sendQueueMsg('testQueue', '123', (error) => {
//     console.log(error)
// })
// mq = null

module.exports = RabbitMQ